package com.motu.monstercity.server.game.verticle;

import com.motu.monstercity.module.common.iap.Iap;
import com.motu.monstercity.module.common.iap.IapGift;
import com.motu.monstercity.module.common.network.ResponseUtils;
import com.motu.monstercity.module.common.utility.ServerParam;
import com.motu.monstercity.protocol.CmdEnum;
import com.motu.monstercity.protocol.Common.PBIapResult;
import com.motu.monstercity.protocol.CsGameIap.CSGameIapFinishResponse;
import com.motu.monstercity.server.game.logic.GameDataManager;
import com.motu.monstercity.server.game.logic.GameIapManager;
import com.motu.monstercity.server.game.logic.GameUser;
import com.motu.monstercity.server.game.logic.IapManager;
import com.motu.vertx.module.utility.base.PayTypeEnum;
import com.motu.vertx.module.utility.mysql.MysqlClient;
import com.motu.vertx.module.utility.mysql.MysqlTool;
import com.motu.vertx.module.utility.threadpool.DBLogManager;
import com.motu.vertx.module.utility.toolset.TimeUtils;
import com.motu.vertx.module.utility.toolset.Tool;
import io.vertx.core.AbstractVerticle;
import io.vertx.core.Future;
import io.vertx.core.Promise;
import io.vertx.core.json.JsonArray;
import io.vertx.core.json.JsonObject;
import io.vertx.kafka.client.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Properties;

//处理iap订单
public class IapKafkaVerticle extends AbstractVerticle {

	public static final int IAP_STATUS_SUCCESS = 1;//订单成功
	public static final int IAP_STATUS_CANCEL = 99;//取消订单
	public static final int IAP_STATUS_IAPID_NO_EXIST = 101;//iapid不存在
	public static final int IAP_STATUS_MONEY_ERROR = 102;//金额错误
	public static final int IAP_STATUS_PRODUCTID_ERROR = 103;//产品ID错误
	public static final int IAP_STATUS_IAPCLICK_ERROR = 104;//iapclick不存在
	public static final int IAP_STATUS_CALLBACK_ERROR = 105;//重复回调
	public static final int IAP_STATUS_IAPID_NO_EQUALS = 106;//iapid不一致
	public static final int IAP_STATUS_IAPORDER_ERROR = 107;//重复游戏订单号
	public static final int IAP_STATUS_LOCK_TIMEOUT_ERROR = 108;//加锁超时
	public static final int IAP_STATUS_IAPORDER_THIRD_ERROR = 109;//重复第三方订单号
	public static final int IAP_STATUS_IAPORDER_DB_ERROR = 110;//订单入库失败
	public static final int IAP_STATUS_TIME_ERROR = 111;//时间异常
	public static final int IAP_STATUS_EXCEPTION = 999;//异常错误

	//支付类型
	public static final String PAY_TYPE_APPLE = "apple";//苹果支付
	public static final String PAY_TYPE_GOOGLE = "google";//谷歌支付
	public static final String PAY_TYPE_WECHAT = "wechat";//微信支付
	public static final String PAY_TYPE_ALIPAY = "alipay";//支付宝支付
	public static final String PAY_TYPE_CHANGMENG = "changmeng";//畅梦支付
	public static final String PAY_TYPE_XINGHUI = "xinghui";//星辉支付

	public final static Logger logger = LoggerFactory.getLogger(IapKafkaVerticle.class);

	public void start(Promise<Void> future) throws Exception {
		String address = this.config().getString("iap-kafka-address");
		int webid = ServerParam.MOTU_WEBID;
		if (Tool.isEmpty(address) || webid == 0) {
			return;
		}
		logger.info("IapKafkaVerticle start webid = " + webid + " address = " + address);
		Properties config = new Properties();
		config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, address);
		config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
		config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
		config.put(ConsumerConfig.GROUP_ID_CONFIG, "iap_group");
		config.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");// 当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据
		config.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");

		KafkaConsumer<String, String> consumer = KafkaConsumer.create(vertx, config);
		//这里lambda表达式用于接收消息处理结果
		String topicName = "sdk_pay_" + webid;
		consumer.subscribe(topicName, ar -> {
			if (ar.succeeded()) {
				logger.info("subscribed " + topicName + " success thread name = " + Thread.currentThread().getName());
				future.complete();
			} else {
				logger.error("Could not subscribe " + topicName + ar.cause().getMessage());
				future.fail(ar.cause());
			}
		});
		consumer.handler(record -> {
			try {
				logger.info("Processing key=" + record.key() + ",value=" + record.value() +
						",partition=" + record.partition() + ",offset=" + record.offset());

				runRechargeLogic(record.value());
			} catch (Throwable t) {
				logger.error(Tool.getException(t));
			} finally {
				consumer.commit(h -> {
					if (h.failed()) {
						logger.error(" ### " + topicName + " consumer.commit error cause : " + Tool.getException(h.cause()));
					}
				});
			}
		});

	}

	/**
	 * 执行充值到账逻辑
	 *
	 * @param rechargeInfo 充值信息
	 * @return 是否成功到账
	 */
	public static boolean runRechargeLogic(String rechargeInfo) {
		JsonObject obj = new JsonObject(rechargeInfo);
		long userId = obj.getLong("userid");
		int platform = obj.getInteger("platform");
		int iapGiftId = obj.getInteger("iapid");//传的是iapGiftId
		double money = obj.getDouble("money");
		String orderId = obj.getString("orderid");
		String productId = obj.getString("productid");
		String transInfo = obj.getString("transinfo");
		String callback = obj.getString("callback");
		String transactionId = obj.getString("transaction_id");//第三方交易ID
		String paytype = obj.getString("paytype");//支付类型

		//验证iapid
		IapGift iapGift = IapManager.getIapGift(iapGiftId);
		Iap iap = IapManager.getIap(iapGift.getIapId());
		if (iapGift == null || iap == null) {
			saveIapError(userId, platform, iap.getId(), iapGiftId, orderId, IAP_STATUS_IAPID_NO_EXIST, rechargeInfo);
			logger.error("add iap order error check iapId error params = " + rechargeInfo);
			return false;
		}
		//验证金额（微信、支付宝、畅梦、37需要验证金额）
		if (PAY_TYPE_WECHAT.equals(paytype) || PAY_TYPE_ALIPAY.equals(paytype) || PAY_TYPE_CHANGMENG.equals(paytype)) {
			if (Double.doubleToLongBits(money) != Double.doubleToLongBits(iap.getMoney())) {
				saveIapError(userId, platform, iap.getId(), iapGiftId, orderId, IAP_STATUS_MONEY_ERROR, rechargeInfo);
				logger.error("add iap order error check money error params = " + rechargeInfo);
				return false;
			}
		}
		//验证产品ID（苹果/谷歌/华为/鸿蒙需要验证产品ID）
		if (PAY_TYPE_APPLE.equals(paytype) || PAY_TYPE_GOOGLE.equals(paytype) || PayTypeEnum.HUAWEI.value.equals(paytype) || PayTypeEnum.HARMONY.value.equals(paytype)) {
			if (Tool.isEmpty(productId) || !IapManager.checkAppProductId(iap.getId(), productId)) {
				saveIapError(userId, platform, iap.getId(), iapGiftId, orderId, IAP_STATUS_PRODUCTID_ERROR, rechargeInfo);
				logger.error("add iap order error check productId error params = " + rechargeInfo);
				return false;
			}
		}
		//验证点击记录
		String sql = "select iap_gift_id,add_time,callback_time,params from iap_click where order_id = ? and user_id = ?;";
		JsonArray params = new JsonArray().add(orderId).add(userId);
		JsonArray array = null;
		try {
			array = Future.await(MysqlTool.queryJsonArrayFromDb(MysqlClient.getGamePool(), sql, params));
		} catch (Throwable e) {
			logger.error("add iap order error select iap_click exception params = " + rechargeInfo);
			logger.error(Tool.getException(e));
			return false;
		}
		if (array == null || array.isEmpty()) {
			saveIapError(userId, platform, iap.getId(), iapGiftId, orderId, IAP_STATUS_IAPCLICK_ERROR, rechargeInfo);
			logger.error("add iap order error iap_click is null params = " + rechargeInfo);
			return false;
		}
		JsonObject objIapclick = array.getJsonObject(0);
		int callbackTime = objIapclick.getInteger("callback_time");
		if (callbackTime > 0) {
			saveIapError(userId, platform, iap.getId(), iapGiftId, orderId, IAP_STATUS_CALLBACK_ERROR, rechargeInfo);
			logger.warn("add iap order error callbackTime exist params = " + rechargeInfo);
			return false;
		}
		int serverIapId = objIapclick.getInteger("iap_gift_id");
		if (serverIapId != iapGiftId) {
			saveIapError(userId, platform, iap.getId(), iapGiftId, orderId, IAP_STATUS_IAPID_NO_EQUALS, rechargeInfo);
			logger.error("add iap order error server iapGiftId = " + serverIapId + " client iapGiftId = " + iapGiftId + " params = " + rechargeInfo);
			return false;
		}
		// 苹果/谷歌充值需要验证票据时间，票据时间要大于iap_click的add_time
		if (PAY_TYPE_APPLE.equals(paytype) || PAY_TYPE_GOOGLE.equals(paytype)) {
			try {
				JsonObject objCallback = new JsonObject(callback);
				long purchase_date_ms;// 票据生成时间（毫秒）
				if (PAY_TYPE_APPLE.equals(paytype)) {
					purchase_date_ms = Long.parseLong(objCallback.getString("purchase_date_ms"));
				} else {
					purchase_date_ms = Long.parseLong(objCallback.getString("purchaseTimeMillis"));
				}
				int add_time = objIapclick.getInteger("add_time");// 充值点击记录生成时间（秒）
				long add_time_ms = (long) add_time * 1000;
				if (purchase_date_ms < add_time_ms) {
					int diffTime = (int) (add_time_ms - purchase_date_ms) / 1000;// 相差时间（秒）
					if (diffTime > ServerParam.RECEIPT_CHECK_TIME) {
						saveIapError(userId, platform, iap.getId(), iapGiftId, orderId, IAP_STATUS_TIME_ERROR, callback);
						logger.error("add iap order error difftime = " + diffTime + " purchase_date_ms = " + purchase_date_ms + " add_time_ms = " + add_time_ms + " params = " + rechargeInfo);
						if (PAY_TYPE_APPLE.equals(paytype)) {
							if (ServerParam.APPSTORE_CHARGE_CHECK_TIME == 1) {
								return false;
							}
						} else {
							if (ServerParam.GOOGLE_CHARGE_CHECK_TIME == 1) {
								return false;
							}
						}
					}
				}
			} catch (Throwable e) {
				logger.error("add iap order check time error orderid = " + orderId + " exception = " + Tool.getException(e));
			}
		}

		//保存订单
		try {
			saveIapOrder(userId, platform, iapGiftId, iap.getBiMoney(), iapGift.getDiamond(), productId, orderId, transInfo, transactionId, paytype);
		} catch (Throwable e) {
			logger.error("add iap order error save order error params = " + rechargeInfo);
			logger.error(Tool.getException(e));
			String errorMsg = e.getMessage();
			int errorCode;
			if (errorMsg.contains("idx_orderid")) {
				errorCode = IAP_STATUS_IAPORDER_ERROR;// 订单重复
			} else if (errorMsg.contains("idx_transactionid_paytype")) {
				errorCode = IAP_STATUS_IAPORDER_THIRD_ERROR;// 第三方订单重复
			} else {
				errorCode = IAP_STATUS_IAPORDER_DB_ERROR;// 入库失败
			}
			saveIapError(userId, platform, iap.getId(), iapGiftId, orderId, errorCode, rechargeInfo + " === error message:" + errorMsg);
			return false;
		}
		//更新点击记录
		updateIapClick(userId, orderId, IAP_STATUS_SUCCESS, callback);
		//获得订单奖励
		try {
			JsonObject jsonObject = new JsonObject();
			jsonObject.put("order_id", orderId);
			jsonObject.put("iap_gift_id", iapGiftId);
			jsonObject.put("trans_info", transInfo);
			jsonObject.put("product_id", productId);
			jsonObject.put("iap_param", objIapclick.getString("params"));// iap购买参数
			jsonObject.put("paytype", paytype);// 支付类型
			CSGameIapFinishResponse.Builder response = CSGameIapFinishResponse.newBuilder();
			PBIapResult.Builder result = null;
			//这里需要加锁处理，防止消费进程刚启动时并发消费同个用户的订单导致数据异常
//                    String lockKey = LockKey.dealOrder(userId);
//                    Lock lock = null;
//                    try {
//                        lock = MainVerticle.getVertLock(lockKey);
			result = GameIapManager.dealOrder(response.getUserBuilder(), userId, jsonObject, false);
			//标记订单已处理
//                        String sql1 = "update iap_order set error_code = 1 where user_id = ? and order_id = ?;";
//                        JsonArray params1 = new JsonArray().add(userId).add(orderId);
//                        Sync.awaitResult(h -> MysqlTool.updateToDb(MainVerticle.instance.jdbcClientGame, sql1, params1, h));

			if (result == null) {
				result = PBIapResult.newBuilder();
			}
			response.setIapResult(result);
			JsonObject obj2 = ResponseUtils.getResponseData(CmdEnum.EnumCmdID.CMD_GAME_IAP_FINISH_VALUE, response.build().toByteArray());
			GameUser.pushIapFinishToUser(userId, obj2);
			return true;
//                    }catch (LockException e){
//                        logger.error("IapKafkaVerticle.dealOrder.lock.timeout record info = " + record.value());
//                        saveIapError(userId, platform, iap.getId(), iapGiftId, orderId, IAP_STATUS_LOCK_TIMEOUT_ERROR, record.value());
//                    }finally {
//                        MainVerticle.releaseVertLock(lock, lockKey);// 释放锁
//                    }
		} catch (Throwable e) {
			logger.error("add iap order error add reward exception params = " + rechargeInfo);
			String errMsg = Tool.getException(e);
			logger.error(errMsg);
			saveIapError(userId, platform, iap.getId(), iapGiftId, orderId, IAP_STATUS_EXCEPTION, rechargeInfo);
			DBLogManager.addServerCrashLog(userId, errMsg);
			return false;
		}
	}

	/***
	 * 订单入库（iaporder表）
	 */
	public static void saveIapOrder(long userId, int platform, int iapGiftId, double money, int diamond, String productId, String orderId, String transInfo, String transactionId, String payType) {
		String sql = "insert into iap_order(user_id,platform,iap_gift_id,money,diamond,product_id,order_id,trans_info,error_code,add_time,transaction_id,pay_type) values(?,?,?,?,?,?,?,?,?,?,?,?);";
		JsonArray params = new JsonArray();
		params.add(userId);
		params.add(platform);
		params.add(iapGiftId);
		params.add(money);
		params.add(diamond);
		params.add(productId);
		params.add(orderId);
		params.add(transInfo);
		params.add(1);// 订单标记为已处理
		params.add(TimeUtils.getCurTime());
		params.add(transactionId);
		params.add(payType);
		Future.await(MysqlTool.updateToDb(MysqlClient.getGamePool(), sql, params));
	}

	/**
	 * 更新iap点击记录
	 */
	public static void updateIapClick(long userId, String orderId, int status, String callback) {
		String sql = "update iap_click set status = ?,callback = ?,callback_time = ? where user_id = ? and order_id = ?;";
		JsonArray params = new JsonArray();
		params.add(status);
		params.add(callback);
		params.add(TimeUtils.getCurTime());
		params.add(userId);
		params.add(orderId);
		MysqlTool.updateToDb(MysqlClient.getGamePool(), sql, params, handler -> {
			if (handler.failed()) {
				logger.error("updateIapClick error cause = {}", Tool.getException(handler.cause()));
			}
		});
	}

	/***
	 * 记录错误订单
	 */
	public static void saveIapError(long userId, int platform, int iapId, int iapGiftId, String orderId, int errCode, String param) {
		String sql = "insert into iap_error(user_id,platform,iap_id,iap_gift_id,order_id,error_code,add_time,param) values(?,?,?,?,?,?,?,?);";
		JsonArray params = new JsonArray();
		params.add(userId);
		params.add(platform);
		params.add(iapId);
		params.add(iapGiftId);
		params.add(orderId);
		params.add(errCode);
		params.add(TimeUtils.getCurTime());
		params.add(param);
		MysqlTool.updateToDb(MysqlClient.getGamePool(), sql, params, handler -> {
			if (handler.failed()) {
				logger.error("saveIapError error cause = {}", Tool.getException(handler.cause()));
			}
		});
	}

}
